package cn.bugstack.ai_autoclassification_qwen.service.impl;

import cn.bugstack.ai_autoclassification_qwen.domain.tProduct;
import cn.bugstack.ai_autoclassification_qwen.repository.tProductRepository;
import cn.bugstack.ai_autoclassification_qwen.service.MultiThreadBatchClassificationService;
import cn.bugstack.ai_autoclassification_qwen.service.TProductClassificationService;
import cn.bugstack.ai_autoclassification_qwen.service.batch.ProductTask;
import cn.bugstack.ai_autoclassification_qwen.service.batch.ProductWorker;
import lombok.RequiredArgsConstructor;
import org.springframework.data.domain.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.*;

@Service
@RequiredArgsConstructor
public class MultiThreadBatchClassificationServiceimpl implements MultiThreadBatchClassificationService {

    private final tProductRepository productRepository;
    private final TProductClassificationService classificationService;

    private static final int THREAD_COUNT = 8;
    private static final int QUEUE_CAPACITY = 5000;

    private final BlockingQueue<ProductTask> queue =
            new ArrayBlockingQueue<>(QUEUE_CAPACITY);

    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public void startMultiThreadClassification() {

        System.out.println("启动多线程分类任务, 线程数=" + THREAD_COUNT);

        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);

        // 启动 worker
        for (int i = 0; i < THREAD_COUNT; i++) {
            executor.submit(new ProductWorker(queue, classificationService));
        }

        int size = 500;

        while (true) {

            // 如果队列里还有很多未处理任务，就等待 worker 消费
            while (queue.size() > size) {
                try {
                    System.out.println("队列未消费完，等待中... queue.size=" + queue.size());
                    Thread.sleep(300); // 等 1 秒
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }

            // 此时队列已空或较少，可以继续扫描数据库
            Pageable pageable = PageRequest.of(0, size, Sort.by("id").ascending());
            Page<tProduct> data =
                    productRepository.findByPredictedCategoryCodeIsNull(pageable);

            List<tProduct> list = data.getContent();
            if (list.isEmpty()) {
                break; // 已无未分类数据 → 结束任务
            }

            System.out.println("扫描到 " + list.size() + " 条未分类数据放入队列...");

            for (tProduct p : list) {
                try {
                    queue.put(new ProductTask(p.getId())); // put 本身会阻塞（安全）
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }

        System.out.println("扫描结束，等待 worker 完成...");
    }
}
